package net.bwie.realtime.jtp.dwd;

import bean.CouponTransaction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Iterator;

public class JtpLogEtIJob {
    /*
    *   实时日志ETL
    *  1.从kafka中读取数据
    *
    * */
    public static void main(String[] args)throws Exception {
        //1.创建执行环境

        StreamExecutionEnvironment  env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 启用 Checkpointing（例如每5秒一次）
        env.enableCheckpointing(5000);
        //2.从kafka中读取数据
        DataStream<String> dataStream = KafkaUtil.consumerKafkaFromLatest(env, "jd-topic");
        //3.数据清洗
        DataStream<String> jsonStream = logCleaned(dataStream);
        DataStream<String> reducedStream=handle(jsonStream);
        //4.打印数据 T
        JdbcUtil.sinkToClickhouseUpsert(reducedStream,"INSERT INTO jtp_app_log.CouponTransaction\n" +
                "(transactionId,\n" +
                " userId,\n" +
                " itemId,\n" +
                " itemName,\n" +
                " itemCategory,\n" +
                " originalPrice,\n" +
                " couponAmount,\n" +
                " paymentAmount,\n" +
                " province,\n" +
                " city,\n" +
                " timestamp,\n" +
                " promotionId)\n" +
                "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)");

        reducedStream.print("reducedStream");

        env.execute("jd-topic");

    }








    private static DataStream<String> logCleaned(DataStream<String> logStream) {
        //脏数据侧边流输出时标记
        OutputTag<String> dirtyTAg = new OutputTag<String>("dirty-log") {
        };
        //脏数据清洗处理
        SingleOutputStreamOperator<String> cleanedStream = logStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    //1.解析json数据
                    JSON.parseObject(value);
                    out.collect(value);
                } catch (Exception e) {
                    ctx.output(dirtyTAg, value);
                }
            }
        });
        //脏数据输出到kafka
        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyTAg);
        KafkaUtil.producerKafka(dirtyStream, "jd-dirty");
        //返回清洗后的数据
        return cleanedStream;
    }
    private static DataStream<String> handle(DataStream<String> kafkaStream) {
        // 1. 将 JSON 字符串转换为 CouponTransaction 对象
        SingleOutputStreamOperator<CouponTransaction> beanStream = kafkaStream.map(new MapFunction<String, CouponTransaction>() {
            @Override
            public CouponTransaction map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return new CouponTransaction(
                        jsonObject.getString("transactionId"),
                        jsonObject.getLong("userId"),
                        jsonObject.getLong("itemId"),
                        jsonObject.getString("itemName"),
                        jsonObject.getString("itemCategory"),
                        jsonObject.getDouble("originalPrice"),
                        jsonObject.getDouble("couponAmount"),
                        jsonObject.getDouble("paymentAmount"),
                        jsonObject.getString("province"),
                        jsonObject.getString("city"),
                        jsonObject.getLong("timestamp"),
                        jsonObject.getString("promotionId")
                );
            }
        });

        // 2. 设置事件时间戳与水位线
        SingleOutputStreamOperator<CouponTransaction> timeStream = beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<CouponTransaction>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp())
                );

        // 3. 按 key 分组（示例以 transactionId 作为 key）
        KeyedStream<CouponTransaction, String> keyedStream = timeStream.keyBy(CouponTransaction::getTransactionId);

        // 4. 开窗操作（每分钟滚动窗口）
        WindowedStream<CouponTransaction, String, TimeWindow> windowStream = keyedStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );

        // 5. 应用窗口函数并转换为 DataStream<String>
        return windowStream.apply(new WindowFunction<CouponTransaction, String, String, TimeWindow>() {
            private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
            @Override
            public void apply(String s, TimeWindow window, Iterable<CouponTransaction> input, Collector<String> out) throws Exception {
                Iterator<CouponTransaction> iterator = input.iterator();
                for (CouponTransaction couponTransaction : input) {
                    String output=couponTransaction.getTransactionId()+','+couponTransaction.getUserId()+','+couponTransaction.getItemId()+','+couponTransaction.getItemName()+','+couponTransaction.getItemCategory()+','+couponTransaction.getOriginalPrice()+','+couponTransaction.getCouponAmount()+','+couponTransaction.getPaymentAmount()+','+couponTransaction.getProvince()+','+couponTransaction.getCity()+','+format.format(couponTransaction.getTimestamp())+','+couponTransaction.getPromotionId();
                    out.collect(output);
                }
            }
        });
    }

}
